/*
 * (c) Copyright 2018 Palantir Technologies Inc. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.palantir.atlasdb.transaction.service;

import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.palantir.atlasdb.coordination.CoordinationService;
import com.palantir.atlasdb.internalschema.InternalSchemaMetadata;
import com.palantir.atlasdb.internalschema.TransactionSchemaManager;
import com.palantir.atlasdb.internalschema.persistence.CoordinationServices;
import com.palantir.atlasdb.keyvalue.api.CheckAndSetCompatibility;
import com.palantir.atlasdb.keyvalue.api.KeyValueService;
import com.palantir.atlasdb.transaction.impl.TransactionConstants;
import com.palantir.atlasdb.transaction.knowledge.TransactionKnowledgeComponents;
import com.palantir.atlasdb.util.MetricsManager;
import com.palantir.atlasdb.util.MetricsManagers;
import com.palantir.timestamp.TimestampService;
import com.palantir.tritium.metrics.registry.DefaultTaggedMetricRegistry;
import com.palantir.tritium.metrics.registry.TaggedMetricRegistry;
import java.util.Map;
import java.util.function.Supplier;

public final class TransactionServices {
    private TransactionServices() {
        // Utility class
    }

    public static TransactionService createTransactionService(
            KeyValueService keyValueService,
            TransactionSchemaManager transactionSchemaManager,
            TransactionKnowledgeComponents knowledge) {
        // Should only be used for testing, or in contexts where users are not concerned about metrics
        return createTransactionService(
                keyValueService, transactionSchemaManager, knowledge, new DefaultTaggedMetricRegistry(), () -> false);
    }

    public static TransactionService createTransactionService(
            KeyValueService keyValueService,
            TransactionSchemaManager transactionSchemaManager,
            TransactionKnowledgeComponents knowledge,
            TaggedMetricRegistry metricRegistry,
            Supplier<Boolean> acceptStagingReadsOnVersionThree) {
        CheckAndSetCompatibility compatibility = keyValueService.getCheckAndSetCompatibility();
        if (compatibility.supportsCheckAndSetOperations() && compatibility.supportsDetailOnFailure()) {
            return createSplitKeyTransactionService(
                    keyValueService,
                    transactionSchemaManager,
                    knowledge,
                    metricRegistry,
                    acceptStagingReadsOnVersionThree);
        }
        return createV1TransactionService(keyValueService);
    }

    private static TransactionService createSplitKeyTransactionService(
            KeyValueService keyValueService,
            TransactionSchemaManager transactionSchemaManager,
            TransactionKnowledgeComponents knowledge,
            TaggedMetricRegistry metricRegistry,
            Supplier<Boolean> acceptStagingReadsOnVersionThree) {
        // TODO (jkong): Is there a way to disallow DIRECT -> V2 transaction service in the map?
        return new PreStartHandlingTransactionService(new SplitKeyDelegatingTransactionService<>(
                transactionSchemaManager::getTransactionsSchemaVersion,
                ImmutableMap.of(
                        TransactionConstants.DIRECT_ENCODING_TRANSACTIONS_SCHEMA_VERSION,
                        createV1TransactionService(keyValueService),
                        TransactionConstants.TICKETS_ENCODING_TRANSACTIONS_SCHEMA_VERSION,
                        createV2TransactionService(keyValueService),
                        TransactionConstants.TWO_STAGE_ENCODING_TRANSACTIONS_SCHEMA_VERSION,
                        createV3TransactionService(keyValueService, metricRegistry, acceptStagingReadsOnVersionThree),
                        TransactionConstants.TTS_TRANSACTIONS_SCHEMA_VERSION,
                        createV4TransactionService(
                                keyValueService, knowledge, metricRegistry, acceptStagingReadsOnVersionThree))));
    }

    public static TransactionService createV1TransactionService(KeyValueService keyValueService) {
        return new PreStartHandlingTransactionService(SimpleTransactionService.createV1(keyValueService));
    }

    private static TransactionService createV2TransactionService(KeyValueService keyValueService) {
        return new PreStartHandlingTransactionService(
                WriteBatchingTransactionService.create(SimpleTransactionService.createV2(keyValueService)));
    }

    private static TransactionService createV3TransactionService(
            KeyValueService keyValueService,
            TaggedMetricRegistry metricRegistry,
            Supplier<Boolean> acceptStagingReadsAsCommitted) {
        return new PreStartHandlingTransactionService(WriteBatchingTransactionService.create(
                SimpleTransactionService.createV3(keyValueService, metricRegistry, acceptStagingReadsAsCommitted)));
    }

    private static TransactionService createV4TransactionService(
            KeyValueService keyValueService,
            TransactionKnowledgeComponents knowledge,
            TaggedMetricRegistry metricRegistry,
            Supplier<Boolean> acceptStagingReadsAsCommitted) {
        return new PreStartHandlingTransactionService(SimpleTransactionService.createV4(
                keyValueService, knowledge, metricRegistry, acceptStagingReadsAsCommitted));
    }

    /**
     * This method should only be used to create {@link TransactionService}s for testing, because in production there
     * are intermediate services like the {@link CoordinationService} this creates where metrics or other forms of
     * lifecycle management may be useful.
     */
    public static TransactionService createRaw(
            KeyValueService keyValueService, TimestampService timestampService, boolean initializeAsync) {
        MetricsManager metricsManager = MetricsManagers.createForTests();
        TransactionKnowledgeComponents knowledge =
                TransactionKnowledgeComponents.createForTests(keyValueService, metricsManager.getTaggedRegistry());
        CoordinationService<InternalSchemaMetadata> coordinationService =
                CoordinationServices.createDefault(keyValueService, timestampService, metricsManager, initializeAsync);
        return createTransactionService(keyValueService, new TransactionSchemaManager(coordinationService), knowledge);
    }

    /**
     * Constructs an {@link AsyncTransactionService} such that methods are blocking and return immediate futures.
     *
     * @param transactionService on which to call synchronous requests
     * @return {@link AsyncTransactionService} which delegates to synchronous methods
     */
    public static AsyncTransactionService synchronousAsAsyncTransactionService(TransactionService transactionService) {
        return new AsyncTransactionService() {
            @Override
            @Deprecated
            public ListenableFuture<Long> getAsync(long startTimestamp) {
                return Futures.immediateFuture(transactionService.get(startTimestamp));
            }

            @Override
            @Deprecated
            public ListenableFuture<Map<Long, Long>> getAsync(Iterable<Long> startTimestamps) {
                return Futures.immediateFuture(transactionService.get(startTimestamps));
            }

            @Override
            public ListenableFuture<TransactionStatus> getAsyncV2(long startTimestamp) {
                return transactionService.getAsyncV2(startTimestamp);
            }

            @Override
            public ListenableFuture<Map<Long, TransactionStatus>> getAsyncV2(Iterable<Long> startTimestamps) {
                return transactionService.getAsyncV2(startTimestamps);
            }
        };
    }
}
